package com.yesun.task.pipeline.handler.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yesun.server.mdp.service.client.config.HttpConfig;
import com.yesun.server.mdp.service.client.impl.DefaultClient;
import com.yesun.server.mdp.service.client.impl.http.HttpRequestContext;
import com.yesun.server.mdp.service.client.impl.http.HttpResponseContext;
import com.yesun.task.pipeline.common.Constant;
import com.yesun.task.pipeline.domain.dto.JobDto;
import com.yesun.task.pipeline.domain.dto.TaskDto;
import com.yesun.task.pipeline.domain.po.NewsPO;
import com.yesun.task.pipeline.handler.Executor;
import com.yesun.task.pipeline.handler.Handler;
import com.yesun.task.pipeline.handler.Loader;
import com.yesun.task.pipeline.service.impl.ClickhouseServiceImpl;
import com.yesun.task.pipeline.sink.ClickhouseSink;
import com.yesun.task.pipeline.worker.ParallelWorker;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.BlockingQueue;

//东方财富新闻采集
@Slf4j
public class DfQuickNewsHandler extends Handler {

    static DefaultClient client = new DefaultClient();
    static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
    static SimpleDateFormat sdfDf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    static SimpleDateFormat shortSdf = new SimpleDateFormat("yyyyMMdd");
    static String TODAY = shortSdf.format(Calendar.getInstance().getTime());
    static Loader loader;

    public DfQuickNewsHandler(JobDto jobDto) {
        super(jobDto);
    }

    @Override
    public Loader getLoader() {
        if(loader == null) {
            //实例化
            loader = new Loader() {
                @Override
                public void load(BlockingQueue queue) {
                    log.info("加载东方财富新闻-7*24快讯采集任务...");

                    //7*24小时快讯，https://newsapi.eastmoney.com/kuaixun/v1/getlist_102_ajaxResult_50_1_.html?r=0.6663938231587259&_=1694228635888
                    TaskDto taskDto = new TaskDto();
                    taskDto.setJobName(jobDto.getJobName());
                    taskDto.setName("东方财富-7*24快讯采集任务");
                    taskDto.setTag(Constant.JOB_TAG_DF_QUICK_NEWS);
                    HashMap map = new HashMap<>();
                    map.put("source", "东方财富网");
                    map.put("category", "7*24快讯");
                    map.put("url", "https://newsapi.eastmoney.com/kuaixun/v1/getlist_102_ajaxResult_50_1_.html?r=0.6663938231587259&_=1694228635888");
                    taskDto.setMap(map);


                    //加入MAP_TASK
                    ParallelWorker.MAP_TASK.put(taskDto.getName(), taskDto.getJobName());

                    queue.offer(taskDto);
                    log.info("添加子任务：" + taskDto.getName());
                }
            };
        }
        return loader;
    }

    @Override
    public Executor getExecutor() {
        return new Executor() {
            @Override
            public void execute(TaskDto taskDto) {
                //执行task的采集
                log.info("execute：" + taskDto.getName());
                long start = System.currentTimeMillis();

                ClickhouseServiceImpl clickhouseServiceImpl = (ClickhouseServiceImpl) com.yesun.task.pipeline.common.SpringUtil.APPLICATION_CONTEXT.getBean("clickhouseServiceImpl");
                //从数据库读取当前分类最新记录
                String timestamp = "";
                List<Map<String, Object>> list = clickhouseServiceImpl.query("SELECT publish_at, topic, category, url FROM ods_stock_news WHERE dt>'20230901' AND category='"+taskDto.getMap().get("category")+"' ORDER BY publish_at DESC limit 1");
                if(list != null &&  list.size()>0) {
                    Map<String,Object> map = list.get(0);
                    timestamp = map.get("publish_at").toString();
                    log.info("timestamp:" + timestamp);
                }

                HttpConfig config = new HttpConfig(taskDto.getMap().get("url"), HttpConfig.HTTP_METHOD_GET);
                config.setEncoding("UTF-8");
                HttpRequestContext request = new HttpRequestContext(config);

                try {
                    HttpResponseContext response = (HttpResponseContext) client.execute(request);
                    String content = response.getContentAsString();
                    content = content.substring(content.indexOf("{"), content.lastIndexOf("}")+1);
                    JSONObject objRoot = JSONObject.parseObject(content);
                    JSONArray jsonArray = objRoot.getJSONArray("LivesList");
                    if (jsonArray != null) {
                        //只需要处理时间戳之后的
                        if(StringUtils.isNotBlank(timestamp)){
                            for(int i=0;i<jsonArray.size();i++) {
                                /**
                                 {"id":"202309092842587035",
                                 "newsid":"202309092842587035",
                                 "url_w":"http://finance.eastmoney.com/news/1355,202309092842587035.html",
                                 "url_m":"http://wap.eastmoney.com/3g/news/article,8,355,1,202309092842587035.shtml",
                                 "url_unique":"http://finance.eastmoney.com/a/202309092842587035.html",
                                 "title":"电影《奥本海默》上映11天总票房突破3亿元",
                                 "simtitle":"电影《奥本海默》上映11天总票房突破3亿元",
                                 "digest":"【电影《奥本海默》上映11天总票房突破3亿元】9月9日，据猫眼专业版数据，电影《奥本海默》上映11天，总票房突破3亿元。",
                                 "simdigest":"【电影《奥本海默》上映11天总票房突破3亿元】9月9日，据猫眼专业版数据，电影《奥本海默》上映11天，总票房突破3亿元。",
                                 "image":"","titlestyle":"0","url_pdf":"","type":"3","simtype":"0","simtype_zh":"","topic":[],"simspecial":[],"simtopic":[],"column":"100,102,105","editor_id":"","editor_name":"","lasteditor_name":"",
                                 "showtime":"2023-09-09 10:51:59","ordertime":"2023-09-09 10:51:59","commentnum":"0","newstype":"1"},
                                **/
                                JSONObject obj = (JSONObject) jsonArray.get(i);
                                Date dtNew = sdfDf.parse(obj.get("showtime").toString());
                                Date dtTimestamp = sdfDf.parse(timestamp);
                                if(dtNew.compareTo(dtTimestamp)>0) {
                                    //表示，新的新闻
                                    //写入到缓存队列，等待批量写入到ck
                                    NewsPO newsPO = new NewsPO();
                                    newsPO.setDt(TODAY);
                                    newsPO.setCategory(taskDto.getMap().get("category").toString());
                                    newsPO.setTopic(obj.get("title").toString());
                                    newsPO.setSummary(obj.get("digest").toString());
                                    newsPO.setPublishAt(obj.get("showtime").toString());
                                    newsPO.setUrl(obj.get("url_w").toString());
                                    newsPO.setSource(taskDto.getMap().get("source").toString());
                                    ClickhouseSink.QUEUE_NEWS.offer(newsPO);
                                } else {
//                                            log.info(obj.get("title").toString() + "，已存在！！！！！");
                                }
                            }
                        } else {
                            for(int i=0;i<jsonArray.size();i++) {
                                JSONObject obj = (JSONObject) jsonArray.get(i);
                                //写入到缓存队列，等待批量写入到ck
                                NewsPO newsPO = new NewsPO();
                                newsPO.setDt(TODAY);
                                newsPO.setCategory(taskDto.getMap().get("category").toString());
                                newsPO.setTopic(obj.get("title").toString());
                                newsPO.setSummary(obj.get("digest").toString());
                                newsPO.setPublishAt(obj.get("showtime").toString());
                                newsPO.setUrl(obj.get("url_w").toString());
                                newsPO.setSource(taskDto.getMap().get("source").toString());
                                ClickhouseSink.QUEUE_NEWS.offer(newsPO);
                            }
                        }
                    }
                }catch (Exception e) {
                    e.printStackTrace();
                }

            }
        };
    }
}
